package drds.binlog.parse.inbound.mysql.dbsync;

import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import drds.binlog.common.position.EntryPosition;
import drds.binlog.parse.driver.packets.server.FieldPacket;
import drds.binlog.parse.driver.packets.server.ResultSetPacket;
import drds.binlog.parse.exception.BinlogParseException;
import drds.binlog.parse.inbound.FieldMetaData;
import drds.binlog.parse.inbound.TableMetaData;
import drds.binlog.parse.inbound.mysql.MysqlConnection;
import drds.binlog.parse.inbound.mysql.ddl.DruidDdlParser;
import drds.binlog.parse.inbound.mysql.tsdb.DatabaseTableMetaData;
import drds.binlog.parse.inbound.mysql.tsdb.MemoryTableMeta;
import drds.binlog.parse.inbound.mysql.tsdb.TableMetaTSDB;
import org.apache.commons.lang.StringUtils;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 处理table meta解析和缓存
 *
 * @author jianghang 2013-1-17 下午10:15:16
 * @version 1.0.0
 */
public class TableMetaDataCache
{

    public static final String COLUMN_NAME = "COLUMN_NAME";
    public static final String COLUMN_TYPE = "COLUMN_TYPE";
    public static final String IS_NULLABLE = "IS_NULLABLE";
    public static final String COLUMN_KEY = "COLUMN_KEY";
    public static final String COLUMN_DEFAULT = "COLUMN_DEFAULT";
    public static final String EXTRA = "EXTRA";
    private MysqlConnection connection;
    private boolean isOnRDS = false;

    private TableMetaTSDB tableMetaTSDB;
    // 第一层tableId,第二层schema.table,解决tableId重复，对应多张表
    private LoadingCache<String, TableMetaData> tableMetaDB;

    public TableMetaDataCache(MysqlConnection con, TableMetaTSDB tableMetaTSDB)
    {
        this.connection = con;
        this.tableMetaTSDB = tableMetaTSDB;
        // 如果持久存储的表结构为空，从db里面获取下
        if (tableMetaTSDB == null)
        {
            this.tableMetaDB = CacheBuilder.newBuilder().build(new CacheLoader<String, TableMetaData>()
            {

                @Override
                public TableMetaData load(String name) throws Exception
                {
                    try
                    {
                        return getTableMetaByDB(name);
                    } catch (Throwable e)
                    {
                        // 尝试做一次retry操作
                        try
                        {
                            connection.reconnect();
                            return getTableMetaByDB(name);
                        } catch (IOException e1)
                        {
                            throw new BinlogParseException("fetch failed by table metadata:" + name, e1);
                        }
                    }
                }

            });
        }

        try
        {
            ResultSetPacket packet = connection.query("show global variables  like 'rds\\_%'");
            if (packet.getFieldValueList().size() > 0)
            {
                isOnRDS = true;
            }
        } catch (IOException e)
        {
        }
    }

    public static List<FieldMetaData> parseTableMeta(String schema, String table, ResultSetPacket packet)
    {
        if (packet.getFieldValueList().size() > 1)
        {
            String createDDL = packet.getFieldValueList().get(1);
            MemoryTableMeta memoryTableMeta = new MemoryTableMeta();
            memoryTableMeta.apply(DatabaseTableMetaData.INIT_POSITION, schema, createDDL, null);
            TableMetaData tableMetaData = memoryTableMeta.find(schema, table);
            return tableMetaData.getFieldMetaDataList();
        } else
        {
            return new ArrayList<FieldMetaData>();
        }
    }

    /**
     * 处理desc table的结果
     */
    public static List<FieldMetaData> parseTableMetaByDesc(ResultSetPacket packet)
    {
        Map<String, Integer> nameMaps = new HashMap<String, Integer>(6, 1f);
        int index = 0;
        for (FieldPacket fieldPacket : packet.getFieldDescriptorList())
        {
            nameMaps.put(fieldPacket.getOriginalColumnName(), index++);
        }

        int size = packet.getFieldDescriptorList().size();
        int count = packet.getFieldValueList().size() / packet.getFieldDescriptorList().size();
        List<FieldMetaData> result = new ArrayList<FieldMetaData>();
        for (int i = 0; i < count; i++)
        {
            FieldMetaData meta = new FieldMetaData();
            // 做一个优化，使用String.intern()，共享String对象，减少内存使用
            meta.setColumnName(packet.getFieldValueList().get(nameMaps.get(COLUMN_NAME) + i * size).intern());
            meta.setColumnType(packet.getFieldValueList().get(nameMaps.get(COLUMN_TYPE) + i * size));
            meta.setNullable(StringUtils.equalsIgnoreCase(packet.getFieldValueList().get(nameMaps.get(IS_NULLABLE) + i
                            * size),
                    "YES"));
            meta.setPrimaryKey("PRI".equalsIgnoreCase(packet.getFieldValueList().get(nameMaps.get(COLUMN_KEY) + i * size)));
            meta.setUniqueKey("UNI".equalsIgnoreCase(packet.getFieldValueList().get(nameMaps.get(COLUMN_KEY) + i * size)));
            // 特殊处理引号
            meta.setDefaultValue(DruidDdlParser.unescapeQuotaName(packet.getFieldValueList()
                    .get(nameMaps.get(COLUMN_DEFAULT) + i * size)));
            meta.setExtra(packet.getFieldValueList().get(nameMaps.get(EXTRA) + i * size));

            result.add(meta);
        }

        return result;
    }

    private TableMetaData getTableMetaByDB(String fullname) throws IOException
    {
        try
        {
            ResultSetPacket packet = connection.query("show create table " + fullname);
            String[] names = StringUtils.split(fullname, "`.`");
            String schema = names[0];
            String table = names[1].substring(0, names[1].length());
            return new TableMetaData(schema, table, parseTableMeta(schema, table, packet));
        } catch (Throwable e)
        { // fallback to desc table
            ResultSetPacket packet = connection.query("desc " + fullname);
            String[] names = StringUtils.split(fullname, "`.`");
            String schema = names[0];
            String table = names[1].substring(0, names[1].length());
            return new TableMetaData(schema, table, parseTableMetaByDesc(packet));
        }
    }

    public TableMetaData getTableMeta(String schema, String table)
    {
        return getTableMeta(schema, table, true);
    }

    public TableMetaData getTableMeta(String schema, String table, boolean useCache)
    {
        if (!useCache)
        {
            tableMetaDB.invalidate(getFullName(schema, table));
        }

        return tableMetaDB.getUnchecked(getFullName(schema, table));
    }

    public TableMetaData getTableMeta(String schema, String table, EntryPosition position)
    {
        return getTableMeta(schema, table, true, position);
    }

    public TableMetaData getTableMeta(String schema, String table, boolean useCache, EntryPosition position)
    {
        TableMetaData tableMetaData = null;
        if (tableMetaTSDB != null)
        {
            tableMetaData = tableMetaTSDB.find(schema, table);
            if (tableMetaData == null)
            {
                // 因为条件变化，可能第一次的tableMeta没取到，需要从db获取一次，并记录到snapshot中
                String fullName = getFullName(schema, table);
                try
                {
                    ResultSetPacket packet = connection.query("show create table " + fullName);
                    String createDDL = null;
                    if (packet.getFieldValueList().size() > 0)
                    {
                        createDDL = packet.getFieldValueList().get(1);
                    }
                    // 强制覆盖掉内存值
                    tableMetaTSDB.apply(position, schema, createDDL, "first");
                    tableMetaData = tableMetaTSDB.find(schema, table);
                } catch (IOException e)
                {
                    throw new BinlogParseException("fetch failed by table metadata:" + fullName, e);
                }
            }
            return tableMetaData;
        } else
        {
            if (!useCache)
            {
                tableMetaDB.invalidate(getFullName(schema, table));
            }

            return tableMetaDB.getUnchecked(getFullName(schema, table));
        }
    }

    public void clearTableMeta(String schema, String table)
    {
        if (tableMetaTSDB != null)
        {
            // tsdb不需要做,会基于ddl sql自动清理
        } else
        {
            tableMetaDB.invalidate(getFullName(schema, table));
        }
    }

    public void clearTableMetaWithSchemaName(String schema)
    {
        if (tableMetaTSDB != null)
        {
            // tsdb不需要做,会基于ddl sql自动清理
        } else
        {
            for (String name : tableMetaDB.asMap().keySet())
            {
                if (StringUtils.startsWithIgnoreCase(name, schema + "."))
                {
                    // removeNames.add(name);
                    tableMetaDB.invalidate(name);
                }
            }
        }
    }

    public void clearTableMeta()
    {
        if (tableMetaTSDB != null)
        {
            // tsdb不需要做,会基于ddl sql自动清理
        } else
        {
            tableMetaDB.invalidateAll();
        }
    }

    /**
     * 更新一下本地的表结构内存
     *
     * @param position
     * @param schema
     * @param ddl
     * @return
     */
    public boolean apply(EntryPosition position, String schema, String ddl, String extra)
    {
        if (tableMetaTSDB != null)
        {
            return tableMetaTSDB.apply(position, schema, ddl, extra);
        } else
        {
            // ignore
            return true;
        }
    }

    private String getFullName(String schema, String table)
    {
        StringBuilder builder = new StringBuilder();
        return builder.append('`')
                .append(schema)
                .append('`')
                .append('.')
                .append('`')
                .append(table)
                .append('`')
                .toString();
    }

    public boolean isOnRDS()
    {
        return isOnRDS;
    }

    public void setOnRDS(boolean isOnRDS)
    {
        this.isOnRDS = isOnRDS;
    }

}
